package net.oschina.simpleplus.erpc.producer;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import net.oschina.simpleplus.erpc.Launcher;
import net.oschina.simpleplus.erpc.core.ERpcRequestEntity;
import net.oschina.simpleplus.erpc.core.ERpcResp;
import net.oschina.simpleplus.erpc.core.ERpcResultEntity;
import net.oschina.simpleplus.erpc.support.ERpcMessageEncoder;

public class ERpcClient implements Launcher {
	
	private static final Logger logger = LoggerFactory.getLogger(ERpcClient.class);

	private Channel channel;

	private EventLoopGroup group = new NioEventLoopGroup();

	private ERpcResultCache eRpcResultCache;

	@Override
	public void start(String host, int port) throws Exception {
		eRpcResultCache = new ERpcResultCache();
		Bootstrap b = new Bootstrap();
		b.group(group).channel(NioSocketChannel.class);
		b.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4),
						new ByteToMessageDecoder() {

							@Override
							protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
									throws Exception {
								ERpcResultEntity entity = new ERpcResultEntity();
								byte[] bs = new byte[in.readableBytes()];
								in.readBytes(bs, in.readerIndex(), in.readableBytes());
								entity.parseFrom(bs);
								eRpcResultCache.setValue(entity.getMsgId(), entity);
								logger.info("receive {}, {}", entity.getMsgId(), entity.getMsg());
							}
						}, new ERpcMessageEncoder());
			}
		});
		b.option(ChannelOption.SO_KEEPALIVE, true);
		channel = b.connect(host, port).sync().channel();
	}

	@Override
	public void shutdown() {
		group.shutdownGracefully();
	}

	@Override
	public ERpcResp write(Object value, Class<?> returnType) {
		ERpcResp put = eRpcResultCache.put(((ERpcRequestEntity)value).getMsgId(), returnType);
		channel.writeAndFlush(value);
		return put;
	}

}
